package org.timewindow

import org.FlinkStreamApp
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector
import org.diysource.SensorSource

/**
 * description ：侧输出流
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/2/23 11:07
 * modified By ：
 * version:    : 1.0
 */
object SideOutputExample extends FlinkStreamApp {
  override def doSomeThing(environment: StreamExecutionEnvironment) = {
    // 获取源数据
    val source = environment.addSource(new SensorSource)

    // 计算
    val minTempStream = source
      .map(l => (l.id, l.temperature))
      .keyBy(_._1)
      .process(new FreezingMonitor)

    // 获取侧输出流并打印
    minTempStream.getSideOutput(new OutputTag[String]("tmp warning")).print

    // 输出结果
//    minTempStream.print
  }

  // K, I, O
  class FreezingMonitor extends KeyedProcessFunction[String, (String, Double), (String, Double)] {
    // 定义一个侧输出流
    lazy val outputTag = new OutputTag[String]("tmp warning")

    override def processElement(value: (String, Double), ctx: KeyedProcessFunction[String, (String, Double), (String, Double)]#Context, out: Collector[(String, Double)]) = {
      // 如果温度低于 32 度，就将结果写到侧输出流中
      if (value._2 < 32) {
        ctx.output(outputTag, s"id = ${value._2}, 温度低于32度")
      }
      // 不管什么情况下，都将原来的数据往下流写
      out.collect(value)
    }
  }

}
